package org.budo.dubbo.protocol.async.repository.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.utils.Bytes;
import org.budo.dubbo.protocol.async.repository.AsyncRepository;
import org.budo.dubbo.protocol.async.repository.BudoAsyncInvocation;
import org.budo.support.lang.util.StringUtil;

import com.alibaba.dubbo.rpc.Invocation;
import com.alibaba.dubbo.rpc.Invoker;
import com.alibaba.dubbo.rpc.RpcInvocation;

import lombok.extern.slf4j.Slf4j;

/**
 * @author lmw
 */
@Slf4j
public class KafkaAsyncRepository extends AbstractKafkaAsyncRepository {
    @Override
    public void sendMessageWhenInvoke(Invocation invocation) {
        Class<?> interfaceType = invocation.getInvoker().getInterface();
        String destinationName = invocation.getInvoker().getUrl().getParameter(AsyncRepository.DESTINATION_NAME);
        if (StringUtil.isEmpty(destinationName)) {
            destinationName = "budo-async-" + interfaceType.getName();
        }

        KafkaProducer<String, Bytes> kafkaProducer = this.getKafkaProducer(interfaceType, destinationName);

        BudoAsyncInvocation budoAsyncInvocation = BudoAsyncInvocation.fromDubboInvocation(invocation);
        byte[] requestBody = this.serializeAsyncInvocation(budoAsyncInvocation, invocation.getInvoker());
        Bytes bytes = Bytes.wrap(requestBody);

        ProducerRecord<String, Bytes> producerRecord = new ProducerRecord<String, Bytes>(destinationName, bytes);

        kafkaProducer.send(producerRecord);

        if (log.isDebugEnabled()) {
            log.debug("#50 sendMessageWhenInvoke, producerRecord=" + producerRecord + ", invocation=" + invocation);
        }
    }

    @Override
    public Integer popMessageAndInvoke(Invoker invoker) {
        long start = System.currentTimeMillis();

        Class<?> interfaceType = invoker.getInterface();
        String destinationName = invoker.getUrl().getParameter(AsyncRepository.DESTINATION_NAME);
        if (StringUtil.isEmpty(destinationName)) {
            destinationName = "budo-async-" + interfaceType.getName();
        }

        KafkaConsumer<String, Bytes> kafkaConsumer = this.getKafkaConsumer(interfaceType, destinationName);
        ConsumerRecords<String, Bytes> consumerRecords = kafkaConsumer.poll(POLL_TIMEOUT);

        if (log.isDebugEnabled()) {
            log.debug("#66 poll 耗时 " + (System.currentTimeMillis() - start) + ", consumerRecords.count=" + consumerRecords.count() + ", POLL_TIMEOUT=" + POLL_TIMEOUT);
        }

        int messageCount = 0;
        for (ConsumerRecord<String, Bytes> consumerRecord : consumerRecords) {
            byte[] messageBody = consumerRecord.value().get();
            BudoAsyncInvocation budoAsyncInvocation = this.deserializeAsyncInvocation(messageBody, invoker);

            if (log.isDebugEnabled()) {
                log.debug("#58 message received, " + consumerRecord + ", budoAsyncInvocation=" + budoAsyncInvocation);
            }

            // 如果消息延迟就打日志
            this.logMessageDelay(budoAsyncInvocation);

            RpcInvocation invocation = BudoAsyncInvocation.toDubboInvocation(budoAsyncInvocation);
            invoker.invoke(invocation); // 执行方法
            messageCount++;
        }

        // 方法调用成功后，提交确认这一批消息
        this.commitOffset(kafkaConsumer);
        return messageCount;
    }
}